package com.atguigu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.app.func.DateFormatUtil;
import com.atguigu.bean.UserLoginBean;
import com.atguigu.util.MyClickHouseUtil;
import com.atguigu.util.MyKafkaUtil;
import netscape.javascript.JSObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.AllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * 数据流: web/app -> Nginx -> 日志服务器（.log）[node01/node02] -> Flume[node01/node02]  -> Kafka(ODS) -> FlinkApp -> Kafka(DWD) -> FlinkApp ->ClickHouse(DWS)
 * 程序： Mock(lg.sh) -> Flume(f1) -> Kafka(ZK) -> BaseLogApp -> Kafka(ZK) -> DwsUserUserLoginWindow-> ClickHouse(ZK)
 */
public class DwsUserUserLoginWindow {
    public static void main(String[] args) throws Exception {
        // TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.setParallelism(4); // 根据kafka的分区数设置
//
//        env.enableCheckpointing(5*1000L, CheckpointingMode.EXACTLY_ONCE);
//        env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//        env.getCheckpointConfig().setCheckpointTimeout(60 * 1000L);
//        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(1), Time.minutes(1)));
//
//        env.setStateBackend(new HashMapStateBackend());
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://10.20.1.231:8020/ck");
//        System.setProperty("HADOOP_USER_HOME", "hadoop");

        // TODO 2.读取kafka 页面日志主题数据
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_user_user_login_window";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        // TODO 3.转换为JSON数据并过滤数据
        SingleOutputStreamOperator<JSONObject> jsonObjectDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                /**
                 {"common":{"ar":"420000","uid":"578","os":"Android 11.0","ch":"huawei","is_new":"1","md":"Xiaomi 9","mid":"mid_380821","vc":"v2.1.132","ba":"Xiaomi"},
                 "page":{"page_id":"good_list","item":"电视","during_time":2552,"item_type":"keyword","last_page_id":"search"},"ts":1684998153000}
                 **/
                // 获取上一页id
                String uid = jsonObject.getJSONObject("common").getString("uid");

                // 获取page_id
                String lastPageId = jsonObject.getJSONObject("page").getString("last_page_id");

                // 当UID不等于空并且上一跳页面为null或者为“login”才是登录数据
                if (uid != null && (lastPageId == null || "login".equals(lastPageId))) {
                    out.collect(jsonObject);
                }

            }
        });

        // TODO 4.提取事件时间设置水位线
        SingleOutputStreamOperator<JSONObject> jsonObjectWithWmDS = jsonObjectDS.assignTimestampsAndWatermarks(WatermarkStrategy
                .<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                                           @Override
                                           public long extractTimestamp(JSONObject element, long recordTimestamp) {
                                               return element.getLong("ts");
                                           }
                                       }
                ));

        // TODO 5.按uuid分组
        KeyedStream<JSONObject, String> keyedStream = jsonObjectWithWmDS.keyBy(json -> json.getJSONObject("common").getString("uid"));

        // TODO 6.使用状态编程统计7日回流用户和独立用户
        SingleOutputStreamOperator<UserLoginBean> userLoginBeanSingleOutputStream = keyedStream.flatMap(new RichFlatMapFunction<JSONObject, UserLoginBean>() {
            private ValueState<String> lastLoginState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 状态描述器
                ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("login-state", String.class);

                // 初始化
                lastLoginState = getRuntimeContext().getState(stringValueStateDescriptor);

            }

            @Override
            public void flatMap(JSONObject value, Collector<UserLoginBean> out) throws Exception {
                // 获取状态数据以及当前状态日期
                Long ts = value.getLong("ts");
                String curDt = DateFormatUtil.toDate(ts);
                String lastDt = lastLoginState.value();

                // 定义当日独立用户数和七日回流用户数
                long uv = 0L;
                long backUv = 0L;

                if (lastDt == null) {
                    uv = 1L;  // 新用户
                    lastLoginState.update(curDt);
                } else {
                    if (!lastDt.equals(curDt)) {  // 如果末次登录日期不等于当前日期
                        uv = 1L;
                        lastLoginState.update(curDt);
                        if (DateFormatUtil.toTs(curDt) - DateFormatUtil.toTs(lastDt) >= 8 * 24 * 3600 * 1000L) {
                            backUv = 1L;
                        }
                    }
                }

                if (uv != 0L) {
                    out.collect(new UserLoginBean("", "", backUv, uv, ts));
                }

            }
        });

        // TODO 7.开窗聚合
        SingleOutputStreamOperator<UserLoginBean> reduceDS = userLoginBeanSingleOutputStream.windowAll(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<UserLoginBean>() {
                    @Override
                    public UserLoginBean reduce(UserLoginBean value1, UserLoginBean value2) throws Exception {
                        value1.setBackCt(value1.getBackCt() + value2.getBackCt());
                        value1.setUuCt(value1.getUuCt() + value2.getUuCt());
                        return value1;
                    }
                }, new AllWindowFunction<UserLoginBean, UserLoginBean, TimeWindow>() {
                    @Override
                    public void apply(TimeWindow window, Iterable<UserLoginBean> values, Collector<UserLoginBean> out) throws Exception {
                        UserLoginBean next = values.iterator().next();

                        next.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        next.setTs(System.currentTimeMillis());

                        out.collect(next);
                    }
                });

        // TODO 8.将数据写入Clickhouse

        reduceDS.print(">>>>>>>>>>>>>>>>>>>>");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("insert into dws_traffic_page_view_window values(?,?,?,?,?)"));

        // TODO 9.启动任务
        env.execute("DwsUserUserLoginWindow");
    }



}
